package com.chenfan.filestore.mq.task;

import cn.hutool.core.date.DateUtil;
import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.ExcelWriter;
import com.alibaba.excel.context.AnalysisContext;
import com.alibaba.excel.metadata.CellExtra;
import com.alibaba.excel.metadata.data.ReadCellData;
import com.alibaba.excel.read.listener.ReadListener;
import com.alibaba.excel.write.metadata.WriteSheet;
import com.alibaba.fastjson.JSONObject;
import com.chenfan.filestore.common.Constants;
import com.chenfan.filestore.config.FileStoreConfig;
import com.chenfan.filestore.disruptor.handler.TransferBusinessDataHandler;
import com.chenfan.filestore.model.BatchImportTemplate;
import com.chenfan.filestore.model.Task;
import com.chenfan.filestore.model.dto.ParseExcelDTO;
import com.chenfan.filestore.service.BatchImportTemplateService;
import com.chenfan.filestore.service.FileService;
import com.chenfan.filestore.service.TaskService;
import com.chenfan.filestore.utils.FileUtil;
import com.chenfan.filestore.vo.FileResult;
import com.chenfan.filestore.vo.ImportResultVO;
import com.chenfan.infra.context.UserContextHolder;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.io.File;
import java.io.InputStream;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;


/**
 * @author K
 * @date 2023/5/17 1:43 PM
 */
@Service
@Slf4j
@RocketMQMessageListener(topic = Constants.BATCH_IMPORT_TASK_TOPIC, consumeThreadMax = 5, consumerGroup = "${spring.application.name}_import_group")
public class ImportTaskConsumer extends BaseMqConsumer implements RocketMQListener<MessageExt>  {
    public static final String ROW_INDEX = "rowIndex";

    @Autowired
    private TaskService taskService;
    @Autowired
    private FileService fileService;
    @Autowired
    private BatchImportTemplateService batchImportTemplateService;
    @Autowired
    private TransferBusinessDataHandler transferBusinessDataHandler;
    @Autowired
    private FileStoreConfig fileStoreConfig;

    @Override
    public void onMessage(MessageExt messageExt) {
        String body = new String(messageExt.getBody());
        TaskMqMsg message = JSONObject.parseObject(body, TaskMqMsg.class);
        String token = String.valueOf(messageExt.getProperty("token"));
        UserContextHolder.setToken(token);
        log.info("开始消费导入数据:{}", body);

        Long taskId = message.getTaskId();
        Task task = taskService.getById(taskId);
        if (Objects.isNull(task)) {
            log.info("未找到对应的task:{}", taskId);
            return;
        }
        task.setDoTime(new Date());
        task.setStatus(Task.STATUS_EXECUTING);
        taskService.updateById(task);

        try {
            BatchImportTemplate template = batchImportTemplateService.getById(task.getTemplateId());
            InputStream inputStream = fileService.getInputStream(task.getDownFileUrl());
            Map<Integer, String> map = new HashMap<>();
            String newLocalFilePath;
            if (StringUtils.isEmpty(task.getFailedFileUrl())) {

                final String[] failedFileUrl = {task.getFailedFileUrl()};
                String localParentDir = new StringBuffer(fileStoreConfig.getTempPath()).append(FileUtil.getUploadDir()).toString();
                String suffix = org.apache.commons.lang.StringUtils.isEmpty(failedFileUrl[0]) ? "xlsx" : FileUtil.suffix(failedFileUrl[0]);
                newLocalFilePath = new StringBuffer(localParentDir).append(FileUtil.uuid() + ".").append(suffix).toString();
                FileUtil.createParentDir(localParentDir);
            } else {
                newLocalFilePath = task.getFailedFileUrl();
            }

            final boolean[] failFlag = {false};

            final ExcelWriter[] excelWriter = {EasyExcel.write(newLocalFilePath).build()};
            WriteSheet writeSheet = EasyExcel.writerSheet(0, "导入失败数据").build();

            ExecutorService executorService = Executors.newFixedThreadPool(4);
            ExecutorCompletionService<ParseExcelDTO> executorCompletionService = new ExecutorCompletionService<>(executorService);
            List<Future<ParseExcelDTO>> list = new ArrayList<>();
            final long[] totalCount = {0};
            final long[] failedCount = {0};

            // 1. 解析excel
            // 拿到原始导入模板
            String templateUrl = template.getTemplateUrl();
            // 构建表头字段 -> 发送业务数据字段
            Map<String, String> fieldMap = extractExcelField(templateUrl);
            log.info("提取模板字段:{}", JSONObject.toJSONString(fieldMap));

            // 1. 解析excel. 分批拆分excel进行保存
            EasyExcel.read(inputStream, new ReadListener<HashMap<String, String>>() {
                private List<Map<String, Object>> cachedDataList = new ArrayList<>();
                Map<String, Map<String, String>> valuesMap = new HashMap<>();
                Map<String, Object> last = null;

                @Override
                public void onException(Exception exception, AnalysisContext context) {
                    log.error(exception.getMessage());
                }

                @Override
                public void invokeHead(Map<Integer, ReadCellData<?>> headMap, AnalysisContext context) {
                    headMap.entrySet().stream().forEach(key -> map.put(key.getKey(), key.getValue().getStringValue()));
                    List<List<String>> header = headMap.entrySet().stream().map(value -> Collections.singletonList(value.getValue().getStringValue())).collect(Collectors.toList());
                    if (!header.contains(Collections.singletonList("失败原因"))) {
                        header.add(Collections.singletonList("失败原因"));
                    }
                    excelWriter[0] = EasyExcel.write(newLocalFilePath).head(header).build();

                }

                @SneakyThrows
                @Override
                public void invoke(HashMap<String, String> data, AnalysisContext context) {
                    Map<String, String> keyValueMap = new HashMap<>();

                    // 当前的行号
                    Integer rowIndex = context.readRowHolder().getRowIndex();

                    data.entrySet().stream().filter(e -> Objects.nonNull(e.getKey())).forEach(key -> {
                        String v = map.get(key.getKey());
                        if (StringUtils.isEmpty(v) || StringUtils.isEmpty(key.getValue()) || Objects.equals(v, "失败原因")) {
                            return;
                        }
                        keyValueMap.put(v, key.getValue());
                    });
                    keyValueMap.put(ROW_INDEX, rowIndex.toString());
                    Map<String, Object> current = transferToBusinessData(fieldMap, keyValueMap);
                    Map<String, Object> transferMap = new HashMap<>();
                    transferMap.put(ROW_INDEX, keyValueMap.get(ROW_INDEX));
                    transferMap.put("data", current);
                    cachedDataList.add(transferMap);

                    valuesMap.put(rowIndex.toString(), keyValueMap);

                    if (cachedDataList.size() >= template.getBatchSize()) {
                        // 往下继续探索, 直至找不到相同单号的数据
                        String groupField = template.getGroupField();
                        // 上一个不为空, 且当前的分组字段跟下一个不相同
                        if (!StringUtils.isEmpty(groupField)) {
                            if (Objects.nonNull(last) && Objects.equals(current.get(groupField), last.get(groupField))) {
                                save(cachedDataList, valuesMap);
                                // 存储完成清理 list
                                cachedDataList = new ArrayList<>();
                                valuesMap = new HashMap<>();
                            }
                        } else {
                            save(cachedDataList, valuesMap);
                            // 存储完成清理 list
                            cachedDataList = new ArrayList<>();
                            valuesMap = new HashMap<>();
                        }
                    }
                    last = current;
                }

                @Override
                public void extra(CellExtra extra, AnalysisContext context) {

                }

                @SneakyThrows
                @Override
                public void doAfterAllAnalysed(AnalysisContext context) {
                    save(cachedDataList, valuesMap);
                }

                @Override
                public boolean hasNext(AnalysisContext context) {
                    return true;
                }

                private void save(List<Map<String, Object>> cachedDataList, Map<String, Map<String, String>> valuesMap) throws ExecutionException, InterruptedException {
                    if (CollectionUtils.isEmpty(cachedDataList)) {
                        return;
                    }
                    ParseExcelDTO parseExcelDTO = ParseExcelDTO.builder()
                            .batchImportTemplate(template)
                            .task(task).sendToBusinessList(cachedDataList)
                            .rowMap(valuesMap)
                            .build();

                    Future<ParseExcelDTO> future = executorCompletionService.submit(new ExecutionTask(parseExcelDTO, transferBusinessDataHandler));
                    list.add(future);
                }
            }).sheet().doRead();
            executorService.shutdown();


            for (int i = 0; i < list.size(); i++) {
                ParseExcelDTO parseExcelDTO = executorCompletionService.take().get();
                if (!CollectionUtils.isEmpty(parseExcelDTO.getResultList())) {
                    List<List<String>> head = excelWriter[0].writeContext().writeWorkbookHolder().getHead();
                    log.info("写入文件的header:{}, 长度:{}", JSONObject.toJSONString(head), head.size());
                    List<List<Object>> writeFileData = getWriteFileData(parseExcelDTO.getRowMap(), head, parseExcelDTO.getResultList());
                    log.info("写入文件的数据长度:{}", writeFileData.size());
                    failFlag[0] = true;
                    totalCount[0] += parseExcelDTO.getSendToBusinessList().size();
                    failedCount[0] += parseExcelDTO.getResultList().size();
                    excelWriter[0].write(writeFileData, writeSheet);
                }
            }

            // 更新完成时间
            task.setDoneTime(new Date());
            task.setStatus(Task.STATUS_SUCCESS);
            task.setTotalCount(totalCount[0]);
            excelWriter[0].finish();

            if (failFlag[0]) {
                FileResult fileResult = fileService.uploadFile(new File(newLocalFilePath));
                task.setFailedFileUrl(fileResult.getId());
                task.setFailedFileName(String.format("%s%s%s.xlsx", template.getTemplateName(), DateUtil.today(), "上传失败文件"));
                if (totalCount[0] != failedCount[0]) {
                    task.setStatus(Task.PART_SUCCESS);
                } else {
                    task.setStatus(Task.STATUS_FAIL);
                }
            }
            updateTaskFinish(task);
        } catch (Exception e) {
            log.error("消费导入任务失败",e);
            // 执行失败记录更新执行结果
            task.setStatus(Task.STATUS_FAIL);
            task.setStackTrace(e.getMessage());
            updateTaskFinish(task);
        }
    }
    private void updateTaskFinish(Task task) {
        task.setDoneTime(new Date());
        task.setUpdateDate(new Date());
        taskService.updateById(task);
    }


    private List<List<Object>> getWriteFileData(Map<String, Map<String, String>> rowMap, List<List<String>> head,
                                 List<ImportResultVO> resultList) {
        return resultList.stream().map(e -> {
            List<Object> objects = new ArrayList<>();
            Integer rowIndex = e.getRowIndex();
            String failReason = e.getFailReason();

            Map<String, String> dataMap = rowMap.get(rowIndex.toString());

            for (int i = 0; i < head.size() - 1; i++) {
                String key = head.get(i).get(0);
                String value = dataMap.get(key);
                objects.add(value);
            }
            objects.add(failReason);
            return objects;
        }).collect(Collectors.toList());
    }

    private Map<String, Object> transferToBusinessData(Map<String, String> fieldMap, Map<String, String> keyValueMap) {
        // 2. 根据模板文件将excel数据转换为对应的list数据
//        Map<String, Object> transferMap = new HashMap<>();
        Map<String, Object> resultMap = new HashMap<>();
        keyValueMap.entrySet().forEach(e -> {
            String key = e.getKey();
            String k = fieldMap.get(key);
            String value = e.getValue();
            if (Objects.isNull(k)) {
                return;
            } else {
                resultMap.put(k, value);
            }
        });
//        transferMap.put(ROW_INDEX, keyValueMap.get(ROW_INDEX));
//        transferMap.put("data", resultMap);
        return resultMap;
    }

    private Map<String, String> extractExcelField(String templateUrl) {
        Map<String, String> keyValueMap = new HashMap<>();
        Map<Integer, String> map = new HashMap<>();
        InputStream remoteTemplate = fileService.getInputStream(templateUrl);

        EasyExcel.read(remoteTemplate, new ReadListener<HashMap<String, String>>() {
            @Override
            public void onException(Exception exception, AnalysisContext context) throws Exception {
                log.error(exception.getMessage());
            }

            @Override
            public void invokeHead(Map<Integer, ReadCellData<?>> headMap, AnalysisContext context) {
                headMap.entrySet().stream().forEach(key -> map.put(key.getKey(), key.getValue().getStringValue()));
            }

            @Override
            public void invoke(HashMap<String, String> data, AnalysisContext context) {
                data.entrySet().forEach(key -> keyValueMap.put(map.get(key.getKey()), key.getValue().replace("{.", "").replace("}", "")));
            }


            @Override
            public void extra(CellExtra extra, AnalysisContext context) {
            }

            @Override
            public void doAfterAllAnalysed(AnalysisContext context) {
            }

            @Override
            public boolean hasNext(AnalysisContext context) {
                return true;
            }
        }).sheet().headRowNumber(1).doRead();
        return keyValueMap;
    }
}

@Slf4j
class ExecutionTask implements Callable<ParseExcelDTO> {

    private ParseExcelDTO parseExcelDTO;
    private TransferBusinessDataHandler transferBusinessDataHandler;

    public ExecutionTask(ParseExcelDTO parseExcelDTO, TransferBusinessDataHandler transferBusinessDataHandler) {
        this.parseExcelDTO = parseExcelDTO;
        this.transferBusinessDataHandler = transferBusinessDataHandler;
    }
    @Override
    public ParseExcelDTO call() {
        return transferBusinessDataHandler.getList(parseExcelDTO);
    }
}
